package com.example.verticles;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.jdbc.JDBCClient;
import io.vertx.ext.sql.ResultSet;
import io.vertx.ext.sql.SQLConnection;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class WikiDatabaseVerticle extends AbstractVerticle {

    public static final String CONFIG_WIKIDB_JDBC_URL = "wikidb.jdbc.url";
    public static final String CONFIG_WIKIDB_JDBC_DRIVER_CLASS = "wikidb.jdbc.driver_class";
    public static final String CONFIG_WIKIDB_JDBC_MAX_POOL_SIZE = "wikidb.jdbc.max_pool_size";
    public static final String CONFIG_WIKIDB_SQL_QUERIES_RESOURCE_FILE = "wikidb.sqlqueries.resource.file";

    public static final String CONFIG_WIKIDB_QUEUE = "wikidb.queue";

    private static final Logger LOGGER = LoggerFactory.getLogger(WikiDatabaseVerticle.class);

    private final HashMap<SqlQuery, String> sqlQueries = new HashMap<>();

    // tag::start[]
    private JDBCClient dbClient;

    @Override
    public void start(Promise<Void> promise) throws Exception {
        /*
         * Note: this uses blocking APIs, but data is small...
         */
        loadSqlQueries();

        dbClient=JDBCClient.createShared(vertx,new JsonObject()
                .put("url", config().getString(CONFIG_WIKIDB_JDBC_URL, "jdbc:hsqldb:file:/home/liulei/develop/hsqldb/wiki"))
                .put("driver_class", config().getString(CONFIG_WIKIDB_JDBC_DRIVER_CLASS, "org.hsqldb.jdbcDriver"))
                .put("max_pool_size", config().getInteger(CONFIG_WIKIDB_JDBC_MAX_POOL_SIZE, 30)));

        dbClient.getConnection(ar->{
            if (ar.failed()){
                LOGGER.error("Could not open a database connection", ar.cause());
                promise.fail(ar.cause());
            }else {
                SQLConnection connection = ar.result();
                connection.execute(sqlQueries.get(SqlQuery.CREATE_PAGES_TABLE),res->{
                    connection.close();
                    if (res.failed()){
                        LOGGER.error("Database preparation error", res.cause());
                        promise.fail(res.cause());
                    }else {
                        vertx.eventBus()
                                .consumer(config().getString(CONFIG_WIKIDB_QUEUE,CONFIG_WIKIDB_QUEUE),
                                        this::onMessage);
                        promise.complete();
                    }
                });
            }
        });
    }
    // end::start[]

    // tag::loadSqlQueries[]
    private enum SqlQuery{
        CREATE_PAGES_TABLE,
        ALL_PAGES,
        GET_PAGE,
        CREATE_PAGE,
        SAVE_PAGE,
        DELETE_PAGE
    }

    private void loadSqlQueries() throws IOException {
        String queriesFile = getConfigValueByKey(CONFIG_WIKIDB_SQL_QUERIES_RESOURCE_FILE);
        InputStream queriesInputStream = StringUtils.isNoneBlank(queriesFile) ?
                new FileInputStream(queriesFile)
                :getClass().getResourceAsStream("/db-queries.properties");
        Properties properties = new Properties();

        properties.load(queriesInputStream);
        queriesInputStream.close();

        sqlQueries.put(SqlQuery.CREATE_PAGES_TABLE, properties.getProperty("create-pages-table"));
        sqlQueries.put(SqlQuery.ALL_PAGES, properties.getProperty("all-pages"));
        sqlQueries.put(SqlQuery.GET_PAGE, properties.getProperty("get-page"));
        sqlQueries.put(SqlQuery.CREATE_PAGE, properties.getProperty("create-page"));
        sqlQueries.put(SqlQuery.SAVE_PAGE, properties.getProperty("save-page"));
        sqlQueries.put(SqlQuery.DELETE_PAGE, properties.getProperty("delete-page"));

    }
    private String getConfigValueByKey(String key){
        return config().getString(key);
    }
    // end::loadSqlQueries[]

    // tag::onMessage[]
    public enum ErrorCodes {
        NO_ACTION_SPECIFIED,
        BAD_ACTION,
        DB_ERROR
    }

    public void onMessage(Message<JsonObject> message) {

        //1.参数校验
        if (!message.headers().contains("action")) {
            LOGGER.error("No action header specified for message with headers {} and body {}",
                    message.headers(), message.body().encodePrettily());
            message.fail(ErrorCodes.NO_ACTION_SPECIFIED.ordinal(), "No action header specified");
            return;
        }

        String action = message.headers().get("action");
        switch (action) {
            case "all-pages":
                fetchAllPages(message);
                break;
            case "get-page":
                fetchPage(message);
                break;
            case "create-page":
                createPage(message);
                break;
            case "save-page":
                savePage(message);
                break;
            case "delete-page":
                deletePage(message);
                break;
            default:
                message.fail(ErrorCodes.BAD_ACTION.ordinal(), "Bad action: " + action);
        }
    }
    // end::onMessage[]

    // tag::rest[]
    private void deletePage(Message<JsonObject> message) {
        JsonArray id = new JsonArray()
                .add(message.body().getString("id"));
        dbClient.updateWithParams(sqlQueries.get(SqlQuery.DELETE_PAGE),id,res->{
            if (res.failed()){
                reportQueryError(message,res.cause());
            }else {
                message.reply("ok");
            }
        });
    }

    private void savePage(Message<JsonObject> message) {
        JsonObject request = message.body();
        JsonArray requestData = new JsonArray()
                .add(request.getString("markdown"))
                .add(request.getString("id"));
        dbClient.updateWithParams(sqlQueries.get(SqlQuery.SAVE_PAGE),requestData,res->{
            if (res.failed()){
                reportQueryError(message,res.cause());
            }else {
                message.reply("ok");
            }
        });
    }

    private void createPage(Message<JsonObject> message) {
        JsonObject request = message.body();
        JsonArray requestData = new JsonArray()
                .add(request.getString("title"))
                .add(request.getString("markdown"));

        dbClient.updateWithParams(sqlQueries.get(SqlQuery.CREATE_PAGE),requestData,res->{
            if (res.failed()){
                message.reply("ok");
            }else {
                reportQueryError(message,res.cause());
            }
        });
    }

    private void fetchPage(Message<JsonObject> message) {
        String pageName = message.body().getString("page");
        JsonArray params = new JsonArray().add(pageName);
        dbClient.queryWithParams(sqlQueries.get(SqlQuery.GET_PAGE),params,res->{
            if (res.failed()){
                reportQueryError(message,res.cause());
            }else {
                JsonObject response=new JsonObject();
                ResultSet resultSet = res.result();
                if (resultSet.getNumRows()==0){
                    response.put("found",false);
                }else {
                    response.put("found",true);
                    JsonArray row = resultSet.getResults().get(0);
                    response.put("id",row.getInteger(0));
                    response.put("rawContent",row.getString(1));
                }
                message.reply(response);
            }
        });
    }

    private void fetchAllPages(Message<JsonObject> message) {
        LOGGER.info("visit fetchAllPages");
        dbClient.query(sqlQueries.get(SqlQuery.ALL_PAGES),res->{
            if (res.failed()){
                reportQueryError(message,res.cause());
            }else {
                List<String> ids = res.result().getResults()
                        .stream()
                        .map(json -> json.getString(0))
                        .sorted()
                        .collect(Collectors.toList());
                message.reply(new JsonObject().put("pages",ids));
            }
        });
    }

    private void reportQueryError(Message<JsonObject> message, Throwable cause){
        LOGGER.error("Database query error", cause);
        message.fail(ErrorCodes.DB_ERROR.ordinal(), cause.getMessage());
    }
    // end::rest[]
}
